#include <ClientThread.h>

/* Class SessionRefreshThread: This class keep the stablished session alive. If the client
   dies, the session will be closed automatically by the sessionManager. */
class SessionRefreshThread : public IceUtil::Thread, public IceUtil::Monitor<IceUtil::Mutex>
{
public:
  
  SessionRefreshThread(const IceUtil::Time& timeout, const Yafridng::RenderSessionPrx& session) :
    _session(session),
    _timeout(timeout),
    _destroyed(false)
  {
    cout << "Session timeout: " << _timeout << endl;
  }
  
  virtual void
  run()
  {
    Lock sync(*this);
    while(!_destroyed)
      {
	timedWait(_timeout);
	if(!_destroyed)
	  {
	    try
	      {
		cout << "Keeping alive" << endl;
		_session->keepAlive();
		cout << "Back keeping alive" << endl;
	      }
	    catch(const Ice::Exception& ex)
	      {
		cerr << "SessionRefreshThread: " << ex;
		_destroyed = true;
	      }
	  }
      }
  }
  
  void
  destroy()
  {
    Lock sync(*this);
    _destroyed = true;
    notify();
  }
  
private:  
  const Yafridng::RenderSessionPrx _session;
  const IceUtil::Time _timeout;
  bool _destroyed;
};
typedef IceUtil::Handle<SessionRefreshThread> SessionRefreshThreadPtr;

/* Class ResultDispatchThread: This class manages incoming tasks for obtaining result files
   and composing them with the result image. */
class ResultDispatchThread : public IceUtil::Thread
{
public:
  ResultDispatchThread(ResultManagerPtr& resultManager) : 
    _resultManager(resultManager)
  {
  }

  virtual void run()
  {
    bool finish = false;
    while(!finish){
      finish = _resultManager->perform();
      sleep(1);
    }
  }
  
private:
  ResultManagerPtr _resultManager;
};
typedef IceUtil::Handle<ResultDispatchThread> ResultDispatchThreadPtr;

/* Class PingNodesThread: This class makes pings to Nodes to check if they are alive. If they
   are not the zone being rendered by the Node will be stablished to NOT_RENDERED. */
class PingNodesThread : public IceUtil::Thread
{
public:
  PingNodesThread(vector<YafridngRendererPrx> renderers, 
		  YafridngRenderManagerIPtr& renderManager,
		  ResultManagerPtr& resultManager) :
    _renderers(renderers),
    _renderManager(renderManager),
    _resultManager(resultManager)
  {
  }

  virtual void run()
  {
    while(!_renderers.empty())
      {
	vector<YafridngRendererPrx>::iterator p = _renderers.begin();
	while(p != _renderers.end())
	  {
	    //Trying to contact the node
	    try
	      {
		(*p)->ice_ping();
		++p;
	      }

	    //If the node is not up, we cancel its tasks
	    catch(const Ice::Exception& ex)
	      {
		cout << "Exception Timeout" << endl;
		_renderManager->cancelRender((*p)->ice_getIdentity().name);
		p = _renderers.erase(p);
	      }
	  }
	sleep(3);
      }
    
    if(!_resultManager->getDestroy()){
      _resultManager->destroy();
    }
  }

private:
  vector<YafridngRendererPrx> _renderers;
  YafridngRenderManagerIPtr _renderManager;
  ResultManagerPtr _resultManager;
};
typedef IceUtil::Handle<PingNodesThread> PingNodesThreadPtr;

class Renderer : public IceUtil::Shared
{
public:
  Renderer(string fileName, 
	   TZones zones,
	   YafridngRendererPrx& renderer, 
	   int piece,
	   Pieces pieces,
	   int fileSize,
	   FilePrx file,
	   P2PBrokerPrx& broker,
	   YafridngRenderManagerPrx& manager,
	   Controller* c): _fileName(fileName),
			   _zones(zones),
			   _renderer(renderer),
			   _piece(piece),
			   _pieces(pieces),
			   _fileSize(fileSize),
			   _file(file),
			   _broker(broker),
			   _manager(manager),
			   _control(c)
  {
  }
  
  void run(){
    try{
      IceUtil::Time p2pTime = IceUtil::Time::now();
      _renderer->obtainFile(_pieces, _piece, _fileSize, _file, _broker);
      cout << "Client has the whole file" << endl;
      IceUtil::Time finalTime = IceUtil::Time::now() - p2pTime;
      cout << endl << "p2pTime Node: " << _renderer->ice_getIdentity().name << "-> " << finalTime.toSecondsDouble() << endl;
      _control->setStatus(_renderer->ice_getIdentity().name, "RENDERING");
      _renderer->render(_zones, _manager);
      _control->setStatus(_renderer->ice_getIdentity().name, "DONE");
    }
    catch(const Yafridng::RenderingFailedException& ex)
      {
	cout << "Exception while rendering:" << ex << endl;
	exit(1);
      }
    catch(const Ice::Exception& ex)
      {
      }
  }

private:
  string _fileName;
  TZones _zones;
  YafridngRendererPrx _renderer;
  int _fileSize;
  FilePrx _file;
  int _piece;
  Pieces _pieces;
  P2PBrokerPrx _broker;
  YafridngRenderManagerPrx _manager;
  Controller* _control;
};
typedef IceUtil::Handle<Renderer> RendererPtr;

class RenderQueue : public IceUtil::Monitor<IceUtil::Mutex>, public IceUtil::Shared
{
public:

  RenderQueue(int);
  
  void add(const RendererPtr& renderer);
  void waitForComplete();
  void destroy();
  RendererPtr get();
  
private:
  
  list<IceUtil::ThreadPtr> _workers;
  list<RendererPtr> _work;
  bool _destroy;
};

typedef IceUtil::Handle<RenderQueue> RenderQueuePtr;

class RenderThread : public IceUtil::Thread
{
public:
  
  RenderThread(const RenderQueuePtr& queue) :
    _queue(queue)
  {
  }
  
  virtual void
  run()
  {
    for(;;)
      {
	RendererPtr work = _queue->get();
	if(!work)
	  {
	    return;
	  }
	work->run();
      }
  }
private:
  const RenderQueuePtr _queue;
};

RenderQueue::RenderQueue(int nworkers) :
    _destroy(false)
{
  for(int i = 0; i < nworkers; ++i)
    {
      IceUtil::ThreadPtr t = new RenderThread(this);
      _workers.push_back(t);
      t->start();
    }
}

void
RenderQueue::add(const RendererPtr& renderer)
{
    Lock sync(*this);
    /*IceUtil::ThreadPtr t = new RenderThread(this);
    _workers.push_back(t);
    t->start();*/
    _work.push_back(renderer);
    notify();
}

void
RenderQueue::waitForComplete()
{
  Lock sync(*this);
  while(!_work.empty())
    {
      wait();
    }
}

void
RenderQueue::destroy()
{
    list<IceUtil::ThreadPtr> workers;
    {
	Lock sync(*this);
	workers = _workers;
	_workers.clear();
	_destroy = true;
	notifyAll();
    }

    for(list<IceUtil::ThreadPtr>::const_iterator p = workers.begin(); p != workers.end(); ++p)
    {
	(*p)->getThreadControl().join();
    }
}

RendererPtr
RenderQueue::get()
{
    Lock sync(*this);
    while(_work.empty() && !_destroy)
    {
	wait();
    }

    if(_destroy)
    {
	return 0;
    }

    RendererPtr work = _work.front();
    _work.pop_front();
    notifyAll(); // This is wakes the thread calling waitForComplete().
    return work;
}

ClientThread::ClientThread(appInfo i, Controller *controller, Ice::CommunicatorPtr& c)
{
  info = i;
  control = controller;
  communicator = c;
}

void ClientThread::run()
{
  TZones zones;
  Pieces pieces;
  vector<YafridngRendererPrx> renderers;
  int startIndex = info.fileName.rfind("/", info.fileName.size());
  string fileName = info.fileName.substr(startIndex + 1, info.fileName.size());
  string path;
  string result = fileName.substr(0, fileName.find(".", 0));
  cout << "Scene name: " << fileName << endl;
  cout << "Zones: " << info.nZones << endl;
  cout << "Renderers: " << info.renderers << endl << endl;
  cout << "Scene resolution: " << endl;
  cout << "\twidth: " << info.width << endl;
  cout << "\theigh: " << info.height << endl << endl;
  
  if(info.sceneType == SCENE)
    {
      result += ".png";
      path = "./temp/";
      Color transp(MaxRGB, MaxRGB, MaxRGB, MaxRGB);
      Magick::Image empty_image(Geometry(info.width, info.height), transp);
      empty_image.write(result);
      control->setImage(result);
    }
  
  else
    {
      string mkdir = "mkdir ";
      mkdir += result;
      system(mkdir.c_str());
      path = "./" + result + "/";
    }
  calculateZones(zones);
  
  Ice::RouterPrx defaultRouter = communicator->getDefaultRouter();
  if(!defaultRouter)
    {
      cerr << ": no default router stablished" << endl;
    }
  
  Glacier2::RouterPrx router = Glacier2::RouterPrx::checkedCast(defaultRouter);
  if(!router)
    {
      cerr <<  ": configured router is not a glacier2 router" << endl;
    }

  try{
    router->destroySession();
  }

  catch(Ice::Exception& ex)
    {
      cout << "Session already destroyed" << endl;
    }

  Yafridng::RenderSessionPrx session;
  try{
    session = Yafridng::RenderSessionPrx::uncheckedCast(router->createSessionFromSecureConnection());
  }
  
  catch(const IceGrid::PermissionDeniedException& ex){
    cout << "Permission denied:\n" << ex.reason << endl;
    control->setFinished(true);
  }

  catch(const Glacier2::CannotCreateSessionException& ex){
    cout << "Cannot Create session: " << ex.reason << endl;
    control->setFinished(true);
  }
  
  SessionRefreshThreadPtr refresh =
    new SessionRefreshThread(IceUtil::Time::seconds(router->getSessionTimeout()/2),
			     session);
  refresh->start();
  
  bool end = false;
  int fileSize = getFileSize(info.fileName);
  int num = info.transferSize;
  int allocatedRenderers = 0;
  /* Obtaining resources for rendering */
  while(!end)
    {
      try
	{
	  Ice::ObjectPrx obj = session->create(info.renderEngine, fileName);
	  YafridngRendererPrx renderer = YafridngRendererPrx::checkedCast(obj);
	  renderers.push_back(renderer);
	  allocatedRenderers++;
	  control->setRenderer(renderer->ice_getIdentity().name);
	  if(allocatedRenderers == info.renderers)
	    end = true;
	}
      //If there are not more nodes
      catch(Yafridng::NoMoreNodesException& ex)
	{
	  cout << "No more allocatable rendererFactories" << endl;
	  end = true;
	}
      //If a node is not executing
      catch(Ice::UnknownLocalException& ex)
	{
	  cout << "Not working node." << ex << endl;
	  end = true;
	}
    }
  
  cout << "Allocated Renderers: " << allocatedRenderers << endl;
  RenderQueuePtr queue = new RenderQueue(allocatedRenderers);
  
  //Obtaining category for bidirectional connections
  string category = router->getServerProxy()->ice_getIdentity().category;
  Ice::ObjectAdapterPtr adapter = communicator->createObjectAdapter("YafridngClientAdapter");
  if(!renderers.empty()){
    Identity renderManagerIdentity;
    renderManagerIdentity.name = "ResultManager";
    renderManagerIdentity.category = category;
    ResultManagerPtr resultManager = new ResultManager(result, info.sceneType);
    YafridngRenderManagerIPtr renderManager = new YafridngRenderManagerI(info.fileName,
									 path,
									 info.width,
									 info.height,
									 zones,
									 resultManager,
									 control);
    Ice::ObjectPrx managerobj = adapter->add(renderManager,
					     renderManagerIdentity);
    YafridngRenderManagerPrx manager = YafridngRenderManagerPrx::uncheckedCast(managerobj);
    
    Identity fileStoreIdentity;
    fileStoreIdentity.name = "FileStore";
    fileStoreIdentity.category = category;
    Ice::ObjectPrx fileobj = adapter->add(new FileStoreI(category), 
					  fileStoreIdentity);
    
    FileStorePrx file = FileStorePrx::uncheckedCast(fileobj);
    calculatePieces(fileSize, renderers.size(), pieces);
    Identity p2pBrokerIdentity;
    p2pBrokerIdentity.name = "p2pBroker";
    p2pBrokerIdentity.category = category;
    
    Ice::ObjectPrx broker = adapter->add(new P2PBrokerI(info.fileName, file, pieces, control),
					 p2pBrokerIdentity);
    Yafridng::P2PBrokerPrx p2pBroker = Yafridng::P2PBrokerPrx::uncheckedCast(broker);
    adapter->activate();
    
    //Creating PingNodesThread
    PingNodesThreadPtr pingNodes = new PingNodesThread(renderers, renderManager, resultManager);
    pingNodes->start();
    
    ResultDispatchThreadPtr resultDispatch = new ResultDispatchThread(resultManager);
    resultDispatch->start();
    
    IceUtil::Time start = IceUtil::Time::now();
    
    /* Adding renderers to the queue. A thread will be executed
       for each of the renderers */
    for(int i = 0; i < renderers.size(); i++){
      queue->add(new Renderer(info.fileName, 
			      zones, 
			      renderers.at(i), 
			      i, 
			      pieces,
			      fileSize,
			      file->read(info.fileName, num, pieces.at(i).offset, pieces.at(i).pieceSize),
			      p2pBroker,
			      manager,
			      control));
    }
    queue->waitForComplete();
    queue->destroy();

    resultDispatch->getThreadControl().join();
    
    //Releasing renderers
    for(int i = 0; i < renderers.size(); i++){
      try
	{
	  renderers.at(i)->flush();
	}
      catch(Ice::Exception& ex)
	{
	  //Blank Exception
	}
    }
    
    pingNodes->getThreadControl().join();

    //If scene is an animation, yafridng compose it
    if(info.sceneType == ANIMATION){
      stringstream w, h;
      w << info.width;
      h << info.height;
      string command = "mencoder 'mf://" + fileName.substr(0, fileName.find(".", 0)) +  "/*.png' -mf w=" + w.str() + ":h=" + h.str() + ":fps=25:type=png -of avi -ovc lavc -o videos/"+ fileName.substr(0, fileName.find(".", 0)) + ".avi";
      
      cout << command << endl;
      }

    IceUtil::Time tm = IceUtil::Time::now() - start;
    cout << endl << "FINAL TIME: " << tm.toSecondsDouble() << endl;
  }
  
  else
    cout << "No renderers available. Try it later" << endl;
  
  refresh->destroy();
  refresh->getThreadControl().join();
  session->destroy();
  /**/
  adapter->destroy();
  cout << "Cambiando finished" << endl;
  control->setFinished(true);
  //return 0;
}

void ClientThread::calculatePieces(int fileSize, int nodes, Pieces& out)
{
  int offset = 0;
  int pieceSize;
  
  if(nodes > 1)
    pieceSize = fileSize/nodes;
  else
    pieceSize = fileSize;
  
  for(int i = 0; i < nodes; i++){
    Piece p;
    if((offset + (2*pieceSize)) > fileSize)
      pieceSize += fileSize%nodes;
    p.offset = offset;
    p.pieceSize = pieceSize;
    out.push_back(p);
    offset += pieceSize;
  }
}

int ClientThread::power(int a, int b){
  if(b == 0)
    {
      return 1;
    }
  
  else{
    int result = 1;
    
    for(int i = 0; i < b; i++)
      {
	result = result * a;
      }
    return result;
  }
}
void ClientThread::calculateZones(TZones& zones){
  if(info.sceneType == SCENE){
    int hsize, vsize, x1, y1, x2, y2, id;
    int l = (int) power(info.nZones, 2);
    hsize = info.width/info.nZones;
    vsize = info.height/info.nZones;
    
    cout << "Calculating work units..." << endl;
    cout << "Valor de l: " << l << endl;
    id = 0;
    for(int i = 0; i < info.nZones; i++){
      x1 = 0;
      y1 = vsize * i;
      x2 = x1 + hsize;
      y2 = y1 + vsize;
      if(y2 + vsize > info.height)
	y2 += info.height%l;
      for(int j = 0; j < info.nZones; j++){
	TZone temp;
	temp.id = id;
	temp.frame = 1;
	temp.x1 = x1;
	temp.y1 = y1;
	temp.x2 = x2;
	temp.y2 = y2;
	x1 += hsize;
	x2 += hsize;
	if(x2 + hsize > info.width)
	  x2 += info.width%info.nZones;
	zones.push_back(temp);
	id++;
      }
    }
  }
  else{
    int id = 0;
    int l = info.endFrame - info.startFrame;
    int frame = info.startFrame;
    for(int i = 0; i < l + 1; i++){
      TZone temp;
      temp.id = id;
      temp.frame = frame;
      temp.x1 = 0;
      temp.y1 = 0;
      temp.x2 = info.width;
      temp.y2 = info.height;
      zones.push_back(temp);
      frame++;
      id++;
    }
  }

  cout << "Numero de zonas total: " << zones.size() << endl;
}

int ClientThread::getFileSize(string file){
  FILE *fp;
  int size;
  if((fp = fopen(file.c_str(), "rb")) == NULL){
    cout << "Error Opening the file " << file << endl;
    exit(1);
  }
  fseek(fp, 0, SEEK_END);
  size = ftell(fp);
  cout << "Size of the file: " << size << endl;
  fclose(fp);
  return size;
}
